package com.xb.system.service.impl;

import com.xb.model.domain.UsersInfoV1;
import com.xb.system.constants.BatchCSVConstants;
import com.xb.system.constants.BatchTxtConstantsV1;
import com.xb.system.constants.ThreadPoolManager;
import com.xb.system.service.FileHandler;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.io.*;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;

/**
 * @author lzy
 * @create 2024/11/4 15:45
 */
@Slf4j
@Component
public class TxtFileHandler implements FileHandler {
    @Value("${file.batch.size}")
    private int batchSize;
    @Value("${file.thread.pool.size}")
    private int threadPoolSize;
    @Autowired
    private DynamicDataSplitterServiceImpl splitterService;

//    @Autowired
//    private BusinessEmailBloomFilterConfig businessEmailBloomFilterConfig;


    @Override
    public void handleFile(String file) throws IOException {
//        BloomFilter<String> businessEmailBloomFilter = businessEmailBloomFilterConfig.createBloomFilter();
        ThreadPoolManager.getExecutor(threadPoolSize);
        BlockingQueue<UsersInfoV1> queue = BatchTxtConstantsV1.QUEUE;
        BatchTxtConstantsV1.SIGNAL_COMPLETION.set(false);
        initConsumerThreads();
        try (BufferedReader reader = new BufferedReader(new FileReader(file))) {
            String line;
            while ((line = reader.readLine()) != null) {
                try {
                    UsersInfoV1 usersInfoV1 = splitterService.processLine(line);
                    if (usersInfoV1 != null) {
                        queue.put(usersInfoV1);
                    }
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }

        } catch (IOException e) {
            log.error("处理 CSV 文档时发生错误: {}", e.getMessage());
            throw e;
        }
    }
    private void initConsumerThreads() {
        for (int i = 0; i < threadPoolSize; i++) {
            ThreadPoolManager.submitTask(() -> {
                try {
                    processQueue();
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    log.error("批处理线程被中断", e);
                }
            });
        }
    }
    private void processQueue() throws InterruptedException {
        List<UsersInfoV1> batch = new ArrayList<>(batchSize);
        BlockingQueue<UsersInfoV1> queue = BatchTxtConstantsV1.QUEUE;
        while (true) {
            UsersInfoV1 usersInfoV1 = queue.poll(200, TimeUnit.MILLISECONDS);
            if (usersInfoV1 != null) {
                batch.add(usersInfoV1);
            }
            if (batch.size() >= batchSize || (BatchCSVConstants.SIGNAL_COMPLETION.get() && queue.isEmpty())) {
                    saveTxtBatch(batch);
                    batch.clear();
                if (BatchCSVConstants.SIGNAL_COMPLETION.get() && queue.isEmpty()) {
                    break;
                }
            }
        }
    }
    private void saveTxtBatch(List<UsersInfoV1> batch) {
        splitterService.saveTxtBatch(batch);
    }


}
